package cn.com.netedge.iot.common.mq;

import cn.com.netedge.iot.common.utils.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public abstract class MqMessageListener {
    private static final Logger logger = LoggerFactory.getLogger(MqMessageListener.class);

    private final ExecutorService executorService = Executors.newFixedThreadPool(10);
    protected String queueName;

    private ConnectionFactory connectionFactory;

    public void start() {
        logger.info("queueName:{} message listener is running!", queueName);
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames(queueName); // 设置监听的队列
        container.setMessageListener(new MessageListenerAdapter(this, "receiveMessage"));
        container.start();
    }

    public void receiveMessage(byte[] message) {
        if(message != null && message.length > 0) {
            try {
                String msg = new String(message, "UTF-8");
                executorService.submit(() -> execute( msg));
                logger.debug("receive queue:{} message:{}", queueName, msg);
                return ;
            } catch (Exception e) {
                logger.debug("message decode error:{}", e.getMessage());
            }
        }
        logger.debug("receive queue:{} message's body is empty.", queueName);
    }

    public void receiveMessage(String message) {
        if(StringUtil.isNotEmpty(message)) {
            try {
                executorService.submit(() -> execute( message));
                logger.debug("receive queue:{} message:{}", queueName, message);
                return ;
            } catch (Exception e) {
                logger.debug("message decode error:{}", e.getMessage());
            }
        }
        logger.debug("receive queue:{} message's body is empty.", queueName);
    }

    public abstract void execute(String msg);

    public boolean checkCacheQueue() {
        return true;
    }

    public void setQueueName(String queueName) {
        this.queueName = queueName;
    }

    public void setConnectionFactory(ConnectionFactory connectionFactory) {
        this.connectionFactory = connectionFactory;
    }
}
